package com.zhouxiaoge;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author 周小哥  gqzmy@163.com 2022/8/9 16:09
 */

public class RichMapFunctionTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        DataStreamSource<Event> source = env.fromElements(new Event("周小哥", "/home", 1000L), new Event("王二", "/pay", 2000L), new Event("张三", "/order", 3000L), new Event("李四", "/bobo", 40000L));

        SingleOutputStreamOperator<Long> streamOperator = source.map(new RichMapFunction<Event, Long>() {

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                System.out.println("索引为" + getRuntimeContext().getIndexOfThisSubtask() + "的任务开始");
            }

            @Override
            public Long map(Event value) throws Exception {
                return value.time;
            }

            @Override
            public void close() throws Exception {
                super.close();
                System.out.println("索引为" + getRuntimeContext().getIndexOfThisSubtask() + "的任务结束");
            }
        });

        streamOperator.print();

        env.execute();
    }
}
